package com.nowcoder.community.event;

import com.alibaba.fastjson.JSONObject;
import com.nowcoder.community.contant.Entity;
import com.nowcoder.community.contant.Topic;
import com.nowcoder.community.contant.User;
import com.nowcoder.community.entity.DiscussPost;
import com.nowcoder.community.entity.Event;
import com.nowcoder.community.entity.Message;
import com.nowcoder.community.property.QiniuProperty;
import com.nowcoder.community.service.DisccusPostService;
import com.nowcoder.community.service.ElasticsearchService;
import com.nowcoder.community.service.MessageService;
import com.nowcoder.community.utils.CommunityUtils;
import com.qiniu.common.QiniuException;
import com.qiniu.storage.Configuration;
import com.qiniu.storage.UploadManager;
import com.qiniu.util.Auth;
import com.qiniu.util.StringMap;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.file.*;
import java.security.Policy;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalTime;
import java.time.Period;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalUnit;
import java.util.HashMap;
import java.util.Map;

@Component
@Slf4j
public class EventConsumer {
    @Autowired
    private MessageService messageService;

    @Autowired
    private ElasticsearchService elasticsearchService;

    @Autowired
    private DisccusPostService disccusPostService;

    @Value("${wk.image.storage}")
    private String wkStorage;

    @Value("${wk.image.cmd}")
    private String wkCmd;

    @Autowired
    private QiniuProperty qiniuProperty;

    @Autowired
    private ThreadPoolTaskExecutor executor;

    /***
     * 系统通知
     * */
    @KafkaListener(topics = {Topic.COMMENT, Topic.LIKE, Topic.FOLLOW})
    public void handleEvent(ConsumerRecord record){
        if (record == null || record.value() == null){
            log.error("消息记录不能为 null");
            return;
        }

        Event event = JSONObject.parseObject((String) record.value(), Event.class);
        if (event == null){
            log.error("事件不能为 null");
            return;
        }
        Message message = new Message();
        message.setFromId(User.SYSTEM);
        message.setToId(event.getEntityUserId());
        message.setConversationId(event.getTopic());
        message.setCreateTime(event.getFireTime());

        Map<String, Object> content = new HashMap<>();
        content.put("userId", event.getUserId());
        content.put("entityType", event.getEntityType());
        content.put("entityId", event.getEntityId());
        event.getData().forEach((k, v) -> content.put(k, v));
        message.setContent(JSONObject.toJSONString(content));
        messageService.addMessage(message);
    }

    /***
     * 更新 ES：添加帖子
     * */
    @KafkaListener(topics = {Topic.POST_PUBLISH})
    public void updateES(ConsumerRecord record){
        if (record == null || record.value() == null){
            log.error("消息不能为 null");
            return;
        }
        Event event = JSONObject.parseObject((String) record.value(), Event.class);
        if (event == null){
            log.error("事件不能为 null");
            return;
        }
        if (event.getEntityType() != Entity.DISCUSS_POST){
            log.error("实体对象必须是帖子");
            return;
        }
        DiscussPost post = disccusPostService.findDiscussPostById(event.getEntityId());
        if (post != null){
            elasticsearchService.addDiscussPost(post);
        }else{
            log.error("es更新的帖子不能为 null");
        }
    }

    /***
     * 更新ES：删除帖子
     * */
    @KafkaListener(topics = {Topic.POST_DELETE})
    public void deletePost(ConsumerRecord record){
        if (record == null || record.value() == null){
            log.error("删帖消息不能为空");
        }
        Event event = JSONObject.parseObject((String) record.value(), Event.class);
        elasticsearchService.deleteDiscussPost(event.getEntityId());
    }

    /**
     * 使用 wkhtmltopdf 将网页转为图片
     * */
    @KafkaListener(topics = {Topic.SHARE})
    public void wkhtmltoimage(ConsumerRecord record){

        Event event = JSONObject.parseObject((String) record.value(), Event.class);
        String url = (String) event.getData("url");
        String filename = (String) event.getData("filename");
        String suffix = (String) event.getData("suffix");
        String path = wkStorage + "/" + filename + "."  + suffix;

        String cmd = wkCmd + " --quality 75 " + url + " " + path;
        try {
            /* 因为.exec() 是异步调用的，所以这里需要启动一个线程去检测文件是否已经生成并上传到云服务器 */
            Runtime.getRuntime().exec(cmd);
            executor.submit(new UploadImageToQiniu(filename, suffix));
        } catch (IOException e) {
            log.error("wkhtmltopdf 生成长图失败：" + e.getMessage());
        }

    }


    class UploadImageToQiniu implements Runnable{

        private String filename;
        private String suffix;
        private int uploadCount;
        private boolean success;
        private final int MAX_WAIT_TIME = 30;
        private final int MAX_UPLOAD_COUNT = 3;
        private final Logger log = LoggerFactory.getLogger(UploadImageToQiniu.class);

        public UploadImageToQiniu(String filename, String suffix) {
            this.filename = filename;
            this.suffix = suffix;
            uploadCount = 0;
            success = false;
        }

        @Override
        public void run() {
            LocalTime now = LocalTime.now();
            LocalTime stopTime = now.plusSeconds(MAX_WAIT_TIME);
            Path path = Paths.get(wkStorage, filename + "." + suffix);
            //  七牛云SDK
            // 设置响应
            StringMap policy = new StringMap();
            policy.put("returnBody", CommunityUtils.getJsonString(200));
            // 生成凭证
            Auth auth = Auth.create(qiniuProperty.getKey().getAccessKey(), qiniuProperty.getKey().getSecretKey());
            String uploadToken = auth.uploadToken(qiniuProperty.getShare().getBucketName(), filename, 60, policy);
            // 上传管理器
            UploadManager manager = new UploadManager(new Configuration());

            while (now.isBefore(stopTime) && uploadCount < MAX_UPLOAD_COUNT){
                if (Files.exists(path, LinkOption.NOFOLLOW_LINKS)){
                    // 图片已经生成则尝试上传
                    try {
                        manager.put(path.toString(), filename, uploadToken, null, "image/" + suffix, false);
                        success = true;
                        break;
                    } catch (QiniuException e) {
                        log.error("UploadManager：上传失败：" + e.getMessage());
                    }
                    uploadCount++;
                }

                // 休眠 500 ms 后再次尝试
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                now = LocalTime.now();
            }

            log.info("上传长图到七牛云: " + success + ", path: " + path);

            // 重命名文件，以便定期清理
            boolean rename = false;
            for (int i = 0; rename == false && i < 10; i++) {
                try {
                    Files.move(path, path.resolveSibling(path.getFileName().toString() + "_ok"));
                    rename = true;
                } catch (IOException ex) {
                    // 如果清理功能正在遍历目录，重命名可能会失败
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException exc) {
                        exc.printStackTrace();
                    }
                }
            }
            log.info("重命名文件状态：" + rename);

        }
    }

}

